use crate::{Error, Result};
use chrono::{DateTime, Utc};
use serde::{Deserialize, Serialize};
use std::collections::VecDeque;
use std::sync::Arc;
use tokio::sync::{broadcast, RwLock};
use tracing::{error, info};
use uuid::Uuid;

/// Event system for handling cluster events
pub struct EventSystem {
    events: Arc<RwLock<VecDeque<Event>>>,
    sender: broadcast::Sender<Event>,
    max_events: usize,
}

/// Cluster event
#[derive(Debug, Clone, Serialize, Deserialize)]
pub struct Event {
    pub id: String,
    pub event_type: String,
    pub timestamp: DateTime<Utc>,
    pub node_id: Option<String>,
    pub data: serde_json::Value,
    pub severity: EventSeverity,
}

/// Event severity levels
#[derive(Debug, Clone, Serialize, Deserialize)]
pub enum EventSeverity {
    Info,
    Warning,
    Error,
    Critical,
}

impl EventSystem {
    /// 创建事件系统
    ///
    /// 初始化广播通道与内存事件队列（环形截断），用于查询与订阅。
    pub fn new() -> Self {
        let (sender, _) = broadcast::channel(1000);

        Self {
            events: Arc::new(RwLock::new(VecDeque::new())),
            sender,
            max_events: 10000,
        }
    }

    /// 订阅事件（返回 broadcast 接收端）
    pub async fn subscribe(&self) -> broadcast::Receiver<Event> {
        self.sender.subscribe()
    }

    /// 发出一个通用事件（存储 + 广播）
    pub async fn emit_event(
        &self,
        event_type: &str,
        node_id: Option<&str>,
        data: serde_json::Value,
        severity: EventSeverity,
    ) -> Result<()> {
        let event = Event {
            id: Uuid::new_v4().to_string(),
            event_type: event_type.to_string(),
            timestamp: Utc::now(),
            node_id: node_id.map(|s| s.to_string()),
            data,
            severity,
        };

        // Store event
        {
            let mut events = self.events.write().await;
            events.push_back(event.clone());

            // Limit event history
            while events.len() > self.max_events {
                events.pop_front();
            }
        }

        // Broadcast event
        if let Err(e) = self.sender.send(event.clone()) {
            error!("Failed to broadcast event: {}", e);
        }

        info!("Event emitted: {} - {}", event.event_type, event.id);
        Ok(())
    }

    /// 查询事件（支持按类型 / 节点 / 起始时间过滤，倒序，限制条数）
    pub async fn get_events(
        &self,
        event_type: Option<String>,
        node_id: Option<String>,
        since: Option<String>,
        limit: usize,
    ) -> Result<Vec<serde_json::Value>> {
        let events = self.events.read().await;

        let since_dt = if let Some(since_str) = since {
            Some(
                DateTime::parse_from_rfc3339(&since_str)
                    .map_err(|e| Error::invalid_request(format!("Invalid since timestamp: {}", e)))?
                    .with_timezone(&Utc),
            )
        } else {
            None
        };

        let filtered_events: Vec<serde_json::Value> = events
            .iter()
            .rev() // Most recent first
            .filter(|event| {
                // Filter by event type
                if let Some(ref filter_type) = event_type {
                    if &event.event_type != filter_type {
                        return false;
                    }
                }

                // Filter by node ID
                if let Some(ref filter_node_id) = node_id {
                    if event.node_id.as_ref() != Some(filter_node_id) {
                        return false;
                    }
                }

                // Filter by timestamp
                if let Some(since_dt) = since_dt {
                    if event.timestamp < since_dt {
                        return false;
                    }
                }

                true
            })
            .take(limit)
            .map(|event| serde_json::to_value(event).unwrap())
            .collect();

        Ok(filtered_events)
    }

    /// 获取当前内存事件总数
    pub async fn get_event_count(&self) -> usize {
        self.events.read().await.len()
    }

    // Specific event emitters

    /// 发出集群初始化事件
    pub async fn emit_cluster_initialized(&self, cluster_id: &str) -> Result<()> {
        let data = serde_json::json!({
            "cluster_id": cluster_id
        });

        self.emit_event("cluster_initialized", None, data, EventSeverity::Info)
            .await
    }

    /// 发出加入集群事件
    pub async fn emit_cluster_joined(&self, cluster_id: &str) -> Result<()> {
        let data = serde_json::json!({
            "cluster_id": cluster_id
        });

        self.emit_event("cluster_joined", None, data, EventSeverity::Info)
            .await
    }

    /// 发出 Raft 领导权变更事件
    pub async fn emit_raft_leadership_change(&self, is_leader: bool) -> Result<()> {
        let data = serde_json::json!({
            "is_leader": is_leader
        });

        let severity = if is_leader {
            EventSeverity::Info
        } else {
            EventSeverity::Warning
        };

        self.emit_event("raft_leadership_change", None, data, severity)
            .await
    }

    /// 发出节点新增事件
    pub async fn emit_node_added(&self, node_id: &str) -> Result<()> {
        let data = serde_json::json!({
            "action": "added"
        });

        self.emit_event("node_lifecycle", Some(node_id), data, EventSeverity::Info)
            .await
    }

    /// 发出节点移除事件
    pub async fn emit_node_removed(&self, node_id: &str) -> Result<()> {
        let data = serde_json::json!({
            "action": "removed"
        });

        self.emit_event("node_lifecycle", Some(node_id), data, EventSeverity::Info)
            .await
    }

    /// 发出受管节点新增事件
    pub async fn emit_managed_node_added(&self, node_id: &str) -> Result<()> {
        let data = serde_json::json!({
            "action": "added"
        });

        self.emit_event(
            "managed_node_lifecycle",
            Some(node_id),
            data,
            EventSeverity::Info,
        )
        .await
    }

    /// 发出受管节点移除事件
    pub async fn emit_managed_node_removed(&self, node_id: &str) -> Result<()> {
        let data = serde_json::json!({
            "action": "removed"
        });

        self.emit_event(
            "managed_node_lifecycle",
            Some(node_id),
            data,
            EventSeverity::Info,
        )
        .await
    }

    /// 发出节点健康状态变更事件
    pub async fn emit_node_health_change(&self, node_id: &str, status: &str) -> Result<()> {
        let data = serde_json::json!({
            "status": status,
            "previous_status": "unknown" // In a real implementation, track previous status
        });

        let severity = match status {
            "healthy" => EventSeverity::Info,
            "unhealthy" => EventSeverity::Error,
            "maintenance" => EventSeverity::Warning,
            _ => EventSeverity::Info,
        };

        self.emit_event("node_health_change", Some(node_id), data, severity)
            .await
    }

    /// 发出配置变更事件
    pub async fn emit_config_change(
        &self,
        section: &str,
        changes: serde_json::Value,
    ) -> Result<()> {
        let data = serde_json::json!({
            "section": section,
            "changes": changes
        });

        self.emit_event("config_change", None, data, EventSeverity::Info)
            .await
    }

    /// 发出故障转移开始事件
    pub async fn emit_failover_started(&self, node_id: &str, reason: &str) -> Result<()> {
        let data = serde_json::json!({
            "reason": reason,
            "action": "started"
        });

        self.emit_event("failover", Some(node_id), data, EventSeverity::Warning)
            .await
    }

    /// 发出故障转移完成事件
    pub async fn emit_failover_completed(
        &self,
        node_id: &str,
        success: bool,
        duration_ms: u64,
    ) -> Result<()> {
        let data = serde_json::json!({
            "action": "completed",
            "success": success,
            "duration_ms": duration_ms
        });

        let severity = if success {
            EventSeverity::Info
        } else {
            EventSeverity::Error
        };

        self.emit_event("failover", Some(node_id), data, severity)
            .await
    }

    /// 发出错误事件
    pub async fn emit_error(
        &self,
        error_type: &str,
        message: &str,
        node_id: Option<&str>,
    ) -> Result<()> {
        let data = serde_json::json!({
            "error_type": error_type,
            "message": message
        });

        self.emit_event("error", node_id, data, EventSeverity::Error)
            .await
    }

    /// 发出快照生成事件
    pub async fn emit_snapshot_taken(&self, log_index: u64, size_bytes: u64) -> Result<()> {
        let data = serde_json::json!({
            "log_index": log_index,
            "size_bytes": size_bytes,
            "action": "taken"
        });

        self.emit_event("snapshot", None, data, EventSeverity::Info)
            .await
    }

    /// 发出快照应用事件
    pub async fn emit_snapshot_applied(&self, log_index: u64) -> Result<()> {
        let data = serde_json::json!({
            "log_index": log_index,
            "action": "applied"
        });

        self.emit_event("snapshot", None, data, EventSeverity::Info)
            .await
    }

    /// 发出成员变更事件（Raft membership 改变）
    pub async fn emit_membership_change(
        &self,
        action: &str,
        node_id: &str,
        new_members: Vec<u64>,
    ) -> Result<()> {
        let data = serde_json::json!({
            "action": action,
            "new_members": new_members
        });

        self.emit_event(
            "membership_change",
            Some(node_id),
            data,
            EventSeverity::Info,
        )
        .await
    }

    /// 发出性能指标事件（可用于接入外部分析流）
    pub async fn emit_performance_metrics(&self, metrics: serde_json::Value) -> Result<()> {
        self.emit_event("performance_metrics", None, metrics, EventSeverity::Info)
            .await
    }
}

impl Default for EventSystem {
    fn default() -> Self {
        Self::new()
    }
}

#[cfg(test)]
mod tests {
    use super::*;
    use tokio::time::{sleep, Duration};

    #[tokio::test]
    async fn test_event_system() {
        let event_system = EventSystem::new();

        // Test event emission
        event_system
            .emit_cluster_initialized("test-cluster")
            .await
            .unwrap();

        // Test event retrieval
        let events = event_system.get_events(None, None, None, 10).await.unwrap();
        assert_eq!(events.len(), 1);

        // Test event subscription
        let mut receiver = event_system.subscribe().await;

        // Emit another event
        event_system.emit_node_added("node-1").await.unwrap();

        // Check if event was received
        tokio::select! {
            event = receiver.recv() => {
                assert!(event.is_ok());
                let event = event.unwrap();
                assert_eq!(event.event_type, "node_lifecycle");
            }
            _ = sleep(Duration::from_millis(100)) => {
                panic!("Event not received");
            }
        }
    }

    #[tokio::test]
    async fn test_event_filtering() {
        let event_system = EventSystem::new();

        // Emit different types of events
        event_system.emit_node_added("node-1").await.unwrap();
        event_system.emit_node_removed("node-2").await.unwrap();
        event_system
            .emit_cluster_initialized("test-cluster")
            .await
            .unwrap();

        // Test filtering by event type
        let node_events = event_system
            .get_events(Some("node_lifecycle".to_string()), None, None, 10)
            .await
            .unwrap();
        assert_eq!(node_events.len(), 2);

        // Test filtering by node ID
        let node1_events = event_system
            .get_events(None, Some("node-1".to_string()), None, 10)
            .await
            .unwrap();
        assert_eq!(node1_events.len(), 1);
    }
}
